-module(pmr).
-behaviour(supervisor).
-behaviour(application).

-compile([verbose, report_errors, report_warnings, trace, debug_info]).
-define(MAX_R, 10).
-define(MAX_T, 60).

-export([init/1, start/0, stop/0,map_fun/1, reduce_fun/2]).
-export([restart/0, start/2, stop/1, map_reduce_job/4]).

start() ->
    net_adm:world(),
    application:start(?MODULE).
stop() ->
    application:stop(?MODULE).

restart() ->
    stop(),
    start().

start(_Type, _Args) ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

stop(_State) ->
    ok.

init(_Args) ->
    {ok, {{one_for_one, ?MAX_R, ?MAX_T},
	  [{server, {server, start_link, []},
	    permanent, 10, worker, [server] }]
	  }}.
	  
map_fun({Key, Value}) ->
    List = string:tokens(Value, " "),
    build_result(List, Key).

build_result([], _Key) ->
    [];
build_result(List, Key) ->
    [H|T] = List,
    [{H, Key}|build_result(T, Key)].

reduce_fun(Key, Values) ->
    Len = length(Values),
    if
	Len > 5 ->
	    io:format("~p exists ~p times~n", [Key, length(Values)]);
	true ->
	    ok
    end.

    
map_reduce_job(Mapper, Reducer, Input, Output) ->
    gen_server:cast(server, {newjob, Mapper, Reducer, Input, Output}).
